package com.alibaba.datax.plugin.reader.ossexcelreader;

import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.read.builder.ExcelReaderBuilder;
import io.minio.MinioClient;
import org.apache.commons.io.Charsets;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.InputStream;
import java.nio.charset.Charset;
import java.nio.charset.UnsupportedCharsetException;
import java.util.*;

/**
 * 目前支持的功能较为简单，参数见excelreader介绍，相对excelreader多增加了minio连接信息，json结构见datax-example-ossexcelreader
 * Created by lizhangxue on 24-7-04.
 */
public class OssExcelReader extends Reader {
	public static class Job extends Reader.Job {
		private static final Logger LOG = LoggerFactory.getLogger(Job.class);

		private Configuration originConfig = null;

		private List<String> path = null;

		private List<String> sourceFiles;

		private MinioUtils minioUtils;

		private String bucket;

		@Override
		public void init() {
			this.originConfig = this.getPluginJobConf();
			//初始化minio连接
			this.initMinioUtils();
			//minio参数检验
			this.validateParameter();
		}

		/**
		 * 初始化minio
		 */
		private void initMinioUtils() {
			try {
				String endpoint = this.originConfig.getString(Key.ENDPOINT);
				String accessId = this.originConfig.getString(Key.ACCESSKEYID);
				String accessKey = this.originConfig.getString(Key.ACCESSKEYSECRET);
				MinioClient minioClient = MinioClient.builder().endpoint(endpoint).credentials(accessId, accessKey).build();
				this.minioUtils = new MinioUtils(minioClient);
			} catch (Exception e) {
				throw DataXException.asDataXException(MinioErrorCode.MINIO_CONNECT_FAIL, "minio连接失败");
			}
		}

		private void validateParameter() {
			// 校验桶数组
			bucket = this.originConfig.getString(Key.BUCKET);
			try {
				boolean bucketExists = minioUtils.bucketExists(bucket);
				if (!bucketExists) {
					throw DataXException.asDataXException(MinioErrorCode.BUCKET_NO_EXIST, "所指定的bucket不存在");
				}
			} catch (Exception e) {
				throw DataXException.asDataXException(MinioErrorCode.BUCKET_QUERY_EXCEPTION, "查询bucket异常");
			}

			// path获取
			String pathInString = this.originConfig.getNecessaryValue(Key.PATH, ExcelReaderErrorCode.REQUIRED_VALUE);
			if (StringUtils.isBlank(pathInString)) {
				throw DataXException.asDataXException(ExcelReaderErrorCode.REQUIRED_VALUE, "您需要指定待读取的源目录或文件");
			}
			if (!pathInString.startsWith("[") && !pathInString.endsWith("]")) {
				path = Collections.singletonList(pathInString);
			} else {
				path = this.originConfig.getList(Key.PATH, String.class);
				if (null == path || path.size() == 0) {
					throw DataXException.asDataXException(ExcelReaderErrorCode.REQUIRED_VALUE, "您需要指定待读取的源目录或文件");
				}
			}

			//字符集编码
			String encoding = this.originConfig.getString(
							com.alibaba.datax.plugin.unstructuredstorage.reader.Key.ENCODING,
							com.alibaba.datax.plugin.unstructuredstorage.reader.Constant.DEFAULT_ENCODING);
			if (StringUtils.isBlank(encoding)) {
                this.originConfig.set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.ENCODING,
                                com.alibaba.datax.plugin.unstructuredstorage.reader.Constant.DEFAULT_ENCODING);
			} else {
				try {
					encoding = encoding.trim();
					this.originConfig.set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.ENCODING, encoding);
					Charsets.toCharset(encoding);
				} catch (UnsupportedCharsetException uce) {
					throw DataXException.asDataXException(ExcelReaderErrorCode.ILLEGAL_VALUE,
							String.format("不支持您配置的编码格式 : [%s]", encoding), uce);
				} catch (Exception e) {
					throw DataXException.asDataXException(ExcelReaderErrorCode.CONFIG_INVALID_EXCEPTION,
							String.format("编码配置异常, 请联系我们: %s", e.getMessage()), e);
				}
			}

			// column: 1. index type 2.value type 3.when type is Date, may have
			// format
			List<Configuration> columns = this.originConfig
					.getListConfiguration(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COLUMN);
			// handle ["*"]
			if (null != columns && 1 == columns.size()) {
				String columnsInStr = columns.get(0).toString();
				if ("\"*\"".equals(columnsInStr) || "'*'".equals(columnsInStr)) {
					this.originConfig.set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COLUMN, null);
					columns = null;
				}
			}

			if (null != columns && columns.size() != 0) {
				for (Configuration eachColumnConf : columns) {
					eachColumnConf.getNecessaryValue(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.TYPE,
									ExcelReaderErrorCode.REQUIRED_VALUE);
					Integer columnIndex = eachColumnConf.getInt(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.INDEX);
					String columnValue = eachColumnConf.getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.VALUE);

					if (null == columnIndex && null == columnValue) {
						throw DataXException.asDataXException(ExcelReaderErrorCode.NO_INDEX_VALUE,
								"由于您配置了type, 则至少需要配置 index 或 value");
					}

					if (null != columnIndex && null != columnValue) {
						throw DataXException.asDataXException(ExcelReaderErrorCode.MIXED_INDEX_VALUE,
								"您混合配置了index, value, 每一列同时仅能选择其中一种");
					}
					if (null != columnIndex && columnIndex < 0) {
						throw DataXException.asDataXException(ExcelReaderErrorCode.ILLEGAL_VALUE, String
										.format("index需要大于等于0, 您配置的index为[%s]", columnIndex));
					}
				}
			}
		}

		@Override
		public void prepare() {
			LOG.debug("prepare() begin...");
			this.sourceFiles = this.buildSourceTargets();
			LOG.info(String.format("您即将读取的文件数为: [%s]", this.sourceFiles.size()));
		}

		@Override
		public void post() {
		}

		@Override
		public void destroy() {
		}

		// warn: 如果源目录为空会报错，拖空目录意图=>空文件显示指定此意图
		@Override
		public List<Configuration> split(int adviceNumber) {
			LOG.debug("split() begin...");
			List<Configuration> readerSplitConfigs = new ArrayList<>();

			// warn:每个slice拖且仅拖一个文件,
			// int splitNumber = adviceNumber;
			int splitNumber = this.sourceFiles.size();
            if (0 == splitNumber) {
                throw DataXException.asDataXException(ExcelReaderErrorCode.EMPTY_DIR_EXCEPTION, String
                                .format("未能找到待读取的文件,请确认您的配置项path: %s", this.originConfig.getString(Key.PATH)));
            }

			List<List<String>> splitedSourceFiles = this.splitSourceFiles(
					this.sourceFiles, splitNumber);
			for (List<String> files : splitedSourceFiles) {
				Configuration splitedConfig = this.originConfig.clone();
				splitedConfig.set(Constant.SOURCE_FILES, files);
				readerSplitConfigs.add(splitedConfig);
			}
			LOG.debug("split() ok and end...");
			return readerSplitConfigs;
		}

		// validate the path, path must be a absolute path
		private List<String> buildSourceTargets() {
			// for each path
			Set<String> toBeReadFiles = new HashSet<>();
			for (String eachPath : this.path) {
				// 添加文件
				if (eachPath.endsWith(Constant.XLS_SUFFIX) || eachPath.endsWith(Constant.XLSX_SUFFIX)) { //文件
					boolean exist = minioUtils.isObjectExist(bucket, eachPath);
					if (!exist) {
						throw DataXException.asDataXException(ExcelReaderErrorCode.FILE_NOT_EXISTS, "文件不存在或者路径不对：" + eachPath);
					}
					toBeReadFiles.add(eachPath);
				} else if (eachPath.endsWith(Constant.ALL_FILE)) {//文件夹
					String folder = eachPath.substring(0, eachPath.length() - 1);
					boolean folderExist = minioUtils.isFolderExist(bucket, folder);
					if (!folderExist) {
						throw DataXException.asDataXException(ExcelReaderErrorCode.FOLDER_NOT_EXISTS, "不是文件夹或文件夹路径不对：" + folder);
					}
					//获取文件夹下的所有文件
					Set<String> listObjects = minioUtils.listObjects(bucket, folder);
					if (listObjects == null || listObjects.size() == 0) {
						throw DataXException.asDataXException(ExcelReaderErrorCode.NO_FOUND_EXCEL, "文件夹下未找到excel文件：" + folder);
					}
					toBeReadFiles.addAll(listObjects);
				} else {
					throw DataXException.asDataXException(ExcelReaderErrorCode.FILE_SUFFIX_ERROR, "配置的oss文件名称或者路径错误");
				}
			}
			return Arrays.asList(toBeReadFiles.toArray(new String[0]));
		}


		private <T> List<List<T>> splitSourceFiles(final List<T> sourceList,
				int adviceNumber) {
			List<List<T>> splitedList = new ArrayList<>();
			int averageLength = sourceList.size() / adviceNumber;
			averageLength = averageLength == 0 ? 1 : averageLength;

			for (int begin = 0, end; begin < sourceList.size(); begin = end) {
				end = begin + averageLength;
				if (end > sourceList.size()) {
					end = sourceList.size();
				}
				splitedList.add(sourceList.subList(begin, end));
			}
			return splitedList;
		}

	}

	public static class Task extends Reader.Task {
		private static final Logger LOG = LoggerFactory.getLogger(Task.class);

		private Configuration readerSliceConfig;
		private List<String> sourceFiles;
		private MinioUtils minioUtils;

		@Override
		public void init() {
			this.readerSliceConfig = this.getPluginJobConf();
			this.sourceFiles = this.readerSliceConfig.getList(Constant.SOURCE_FILES, String.class);
			//初始化minio连接
			this.initMinioUtils();
		}

		/**
		 * 初始化minio
		 */
		private void initMinioUtils() {
			try {
				String endpoint = this.readerSliceConfig.getString(Key.ENDPOINT);
				String accessId = this.readerSliceConfig.getString(Key.ACCESSKEYID);
				String accessKey = this.readerSliceConfig.getString(Key.ACCESSKEYSECRET);
				MinioClient minioClient = MinioClient.builder().endpoint(endpoint).credentials(accessId, accessKey).build();
				this.minioUtils = new MinioUtils(minioClient);
			} catch (Exception e) {
				throw DataXException.asDataXException(MinioErrorCode.MINIO_CONNECT_FAIL, "minio连接失败");
			}
		}

		@Override
		public void prepare() {

		}

		@Override
		public void post() {

		}

		@Override
		public void destroy() {

		}

		@Override
		public void startRead(RecordSender recordSender) {
			LOG.info("start read source files...");
			Boolean multipleSheets = this.readerSliceConfig
					.getBool(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.MULTIPLE_SHEET,
							com.alibaba.datax.plugin.unstructuredstorage.reader.Constant.DEFAULT_MULTIPLE_SHEET);
			String bucket = this.readerSliceConfig.getString(Key.BUCKET);
			for (String fileName : this.sourceFiles) {
				LOG.info(String.format("reading file : [%s]", fileName));
				try (InputStream fileInputStream = minioUtils.getObject(bucket, fileName)){
					ExcelNoModelDataListener excelNoModelDataListener = new ExcelNoModelDataListener(fileName,
							this.readerSliceConfig, recordSender, this.getTaskPluginCollector());
					ExcelReaderBuilder excelReaderBuilder = EasyExcel.read(fileInputStream, excelNoModelDataListener);
					// excelReader读取参数设置
					settingExcelReaderParams(excelReaderBuilder, fileName);
					if (multipleSheets) {
						excelReaderBuilder.doReadAll();
					} else {
						excelReaderBuilder.sheet().doRead();
					}
				} catch (Exception e) {
					e.printStackTrace();
					String message = String.format("读取文件失败, 文件名 : [%s]", fileName);
					LOG.error(message);
					throw DataXException.asDataXException(ExcelReaderErrorCode.OPEN_FILE_ERROR, message);
				}
			}
			LOG.debug("end read source files...");
		}

		private void settingExcelReaderParams(ExcelReaderBuilder excelReaderBuilder, String fileName) {
			//字符集编码设置
			String encoding = this.readerSliceConfig.getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.ENCODING);
			if (StringUtils.isNotBlank(encoding)) {
				excelReaderBuilder.charset(Charset.forName(encoding));
			}
			//跳过几行
			Integer skipRows = this.readerSliceConfig.getInt(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.SKIP_ROWS);
			if (skipRows != null) {
				excelReaderBuilder.headRowNumber(skipRows);
				LOG.info(String.format("文件路径为 %s 的文件将跳过表格前 %s 行", fileName, skipRows));
			}
		}

	}
}
